package com.cloud.common.akka;

import akka.actor.UntypedAbstractActor;
import com.cloud.common.context.AppContext;
import com.cloud.common.context.SessionContext;
import com.cloud.common.context.TraceContext;
import com.cloud.common.context.VersionContext;
import lombok.extern.slf4j.Slf4j;

/**
 * actor父类
 *
 * @author fengwenjin
 */
@Slf4j
public abstract class BaseActor extends UntypedAbstractActor {

    private static final ThreadLocal<ActorMessage> messageHolder = new ThreadLocal<>();

    @Override
    public void onReceive(Object message) {
        if (message instanceof ActorMessage == false) {
            log.warn("receive message must ActorMessage");
            return;
        }
        try {
            ActorMessage actorMessage = (ActorMessage) message;
            messageHolder.set(actorMessage);
            TraceContext.set(actorMessage.traceId);
            VersionContext.set(actorMessage.version);
            SessionContext.set(actorMessage.sessionInfo);

            onMessage(actorMessage);
        } finally {
            messageHolder.remove();
            TraceContext.remove();
            VersionContext.remove();
            SessionContext.remove();
        }
    }

    public void tell(Class actor, ActorMessage message, boolean completed) {
        ActorMessage prevMessage = messageHolder.get();
        message.batchId = prevMessage.batchId;
        message.countDownLatch = prevMessage.countDownLatch;
        message.completed = completed;
        AppContext.getBean(ActorManager.class).tell(actor, message);
    }

    public abstract void onMessage(ActorMessage message);
}
